package com.bw.djb

import java.text.SimpleDateFormat

import org.apache.spark.sql.SparkSession

object spark_test01 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder().master("local[*]").appName("sparktest").getOrCreate()


    val score = spark.read.option("sep", ",").schema("student_id int,class_id int,grade int")
      .csv("input\\score_info.txt")
    score.createOrReplaceTempView("score_info")
    val course = spark.read.option("sep", ",").schema("id int,name string,teacher_id int")
      .csv("input\\course_info.txt")
    course.createOrReplaceTempView("course_info")
    val teacher = spark.read.option("sep", ",").schema("id string,name string")
      .csv("input\\teacher_info.txt")

    teacher.createOrReplaceTempView("teacher_info")
    val store = spark.read.option("sep", ",").schema("store_type string,store_name string,store_level string,join_date string," +
      "monthly_sales string,fans string,score string,comment_num string")
      .csv("input\\store_info.txt")
    store.createOrReplaceTempView("store_info")
    val student = spark.read.option("sep", ",").schema("id int,name string,birthday string,sex string,class string,type string")
      .csv("input\\student_info.txt")
    student.createOrReplaceTempView("student_info")


    println("查询出三门科目的平均成绩，并将其代课老师打印到控制台")
    spark.sql(
      """
        |select
        |avg(s.grade) avg_score,t.name
        |from
        |score_info s
        |join course_info c on s.class_id = c.id
        |join teacher_info t on c.teacher_id=t.id
        |group by t.name
        |""".stripMargin)show(100, true) // Show 100 rows & full column value

    println("查询所有学生信息，将性别进行转换：男->man 女->women")
    spark.sql(
      """
        | SELECT
        |  id,
        |  name,
        |  CASE WHEN sex = '男' THEN 'man' ELSE 'women' END AS sex,
        |  birthday,
        |  class,
        |  type
        |FROM
        |  student_info
        |""".stripMargin)show(100, false)

    println("查询出有任意两门成绩低于60的学生信息")
    spark.sql(
      """
        |SELECT
        |*
        |FROM
        |student_info
        |WHERE id IN (
        |SELECT student_id FROM score_info WHERE grade < 60 GROUP BY student_id HAVING COUNT(*) >= 2
        |)
        |""".stripMargin)show(100, false)

    println("查询出语文成绩前3的学生信息及分数")
    spark.sql(
      """
        |SELECT grade,st.name
        |FROM score_info sc
        |join student_info st on sc.student_id = st.id
        |ORDER BY grade DESC
        |LIMIT 3
        |""".stripMargin)show(100, false)

    println("查询出男生三科总分的平均分以及女生三科总分的平均分")
    spark.sql(
      """
        | select
        | sco.class_id,avg(sco.grade) manavg
        | from
        | student_info stu
        | join score_info sco on stu.id = sco.student_id
        | where  sex = "男"
        | group by sco.class_id
        |""".stripMargin).show()

    spark.sql(
      """
        | select
        | sco.class_id,avg(sco.grade) womenavg
        | from
        | student_info stu
        | join score_info sco on stu.id = sco.student_id
        | where  sex = "女"
        | group by sco.class_id
        |""".stripMargin).show()
    println("查询出哪位同学学过的科目数小于3")
    spark.sql(
      """
        |select
        |*
        |from (
        |select
        |stu.id,stu.name,count(sco.class_id) cns
        |from
        |student_info stu
        |join score_info sco
        |on
        |stu.id = sco.student_id
        |group by stu.id,stu.name ) t1 where t1.cns < 3
        |""".stripMargin).show()
    println("查询和编号02教师同姓的同学信息以及总分")
    spark.sql(
      """
        | select
        | name,sum(grade) sum1
        | from (
        | select * from(
        | select * from
        | student_info stu
        | join score_info sco on stu.id = sco.student_id
        | )t1,(
        | select substring(name,0,1) tname1 from teacher_info where id = '1002'
        | ) t2 where substring(t1.name,0,1)=t2.tname1
        | ) t3 group by name
        |""".stripMargin).show()
    println("查询编号03的教师所交科目中有哪些学生本科成绩小于60分")
    spark.sql(
      """
        |select
        |*
        |from student_info stu
        | join score_info sco on stu.id=sco.student_id
        | join course_info cou on sco.class_id=cou.id
        | join teacher_info tea on cou.teacher_id=tea.id
        |where tea.id='1003' and grade <60
        |""".stripMargin).show()


    spark.stop()
  }
}
